package com.ruyuan.engine.core;
import com.ruyuan.engine.datagen.MockDataUtils;
import com.ruyuan.engine.datagen.RuleComplexParamDataGen;
import com.ruyuan.engine.pojo.ClientLog;
import com.ruyuan.engine.pojo.RuleComplexParam;
import com.ruyuan.engine.pojo.RuleEventExecuteParam;
import com.ruyuan.engine.pojo.RuleResult;
import com.ruyuan.engine.service.*;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
/**
 * 需求：最近6个小时的数据查询ListState，超过6个小时的数据查询ClickHouse
 * */
public class RuleEngineProcessDynamic extends KeyedProcessFunction<String, ClientLog, RuleResult> {

    protected static final Logger log = LoggerFactory.getLogger(RuleEngineProcessDynamic.class);

    /**
     * 注入服务
     * */
    private BaseRuleService baseRuleService;
    private CombinationRuleService combinationRuleService;
    private SequenceRuleService sequenceRuleService;
    private CombinationRuleService combinationRuleServiceClickHouseManagerImpl;
    private SequenceRuleService sequenceRuleServiceClickHouseManagerImpl;
    ListState<ClientLog> clientLogListState;
    //动态生成的规则对象
    RuleComplexParam ruleComplexParam;

    @Override
    public void open(Configuration parameters) throws Exception {
        /**
         * clientLogListState只保存6个小时的数据
         * */
        ListStateDescriptor<ClientLog> descriptor = new ListStateDescriptor<ClientLog>("clientLogListState",ClientLog.class);
        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.hours(6)).updateTtlOnCreateAndWrite().build();
        descriptor.enableTimeToLive(stateTtlConfig);
        clientLogListState = getRuntimeContext().getListState(descriptor);
        ruleComplexParam = RuleComplexParamDataGen.mockData();
        baseRuleService = new BaseRuleServiceImpl();
        combinationRuleService = new CombinationRuleServiceImpl(clientLogListState);
        sequenceRuleService = new SequenceRuleServiceImpl(clientLogListState);
        combinationRuleServiceClickHouseManagerImpl = new CombinationRuleServiceClickHouseManagerImpl();
        sequenceRuleServiceClickHouseManagerImpl = new SequenceRuleServiceClickHouseManagerImpl();
    }

    @Override
    public void processElement(ClientLog clientLog, Context ctx, Collector<RuleResult> out) throws Exception {
        Boolean condition1 = false;
        Boolean condition2 = false;
        Boolean condition3 = false;
        clientLogListState.add(clientLog);
        //获取当前系统时间的前6个小时.
        long timeLimit6H = System.currentTimeMillis() - 6 * 60 * 60 * 1000;

        /**
         * 主流程
         * */
        if (ruleComplexParam.getTrigger().getEventId().equals(clientLog.getEventId())) {
            //基本规则条件
            condition1 = baseRuleService.mathRuleByCondition(clientLog, ruleComplexParam);

            /**
             * 组合规则查询：
             *      根据时间跨度来选择不同的查询方式，先把规则中的事件分类处理
             * */
            ArrayList<RuleEventExecuteParam> before6H = new ArrayList<>();
            ArrayList<RuleEventExecuteParam> after6H = new ArrayList<>();

            //遍历组合规则数据放到不同的分组
            /**
             * 需要解决时间范围问题：
             * */
            List<RuleEventExecuteParam> combinationRuleParams = ruleComplexParam.getCombinationRuleParams();
            for (RuleEventExecuteParam ruleEventExecuteParam:combinationRuleParams) {
                if (ruleEventExecuteParam.getEventStartTime() < timeLimit6H) {
                    //之前的数据，查询clickhouse
                    before6H.add(ruleEventExecuteParam);
                }else {
                    after6H.add(ruleEventExecuteParam);
                }
            }

            //查询state
            if (after6H.size() > 0) {
                ruleComplexParam.setCombinationRuleParams(after6H);
                boolean b = combinationRuleService.mathRuleConditionCount(clientLog.getUserNo(), ruleComplexParam);
                if (b) condition2 = b;
            }
            //查询clickhouse
            if (before6H.size() >0) {
                ruleComplexParam.setCombinationRuleParams(before6H);
                boolean b = combinationRuleServiceClickHouseManagerImpl.mathRuleConditionCount(clientLog.getUserNo(), ruleComplexParam);
                if (b) condition2 = b;
            }

            /**
             * 顺序查询的话：1 -  2 -  3
             * 依据第一个规则的时间作为标准
             * */
            List<RuleEventExecuteParam> sequenceRuleParams = ruleComplexParam.getSequenceRuleParams();

            //判断规则的时间是否大于6小时
            if (sequenceRuleParams.size() >0 && sequenceRuleParams.get(0).getEventStartTime() >= timeLimit6H) {
                //在6个小时范围内
                boolean b = sequenceRuleService.mathRuleSequenceCount(ruleComplexParam);
                if (b) condition3 = b;
            }

            if (sequenceRuleParams.size() >0 && sequenceRuleParams.get(0).getEventStartTime() < timeLimit6H) {
                boolean b = sequenceRuleServiceClickHouseManagerImpl.mathRuleSequenceCount(clientLog, ruleComplexParam);
                if (b) condition3 = b;
            }
            if (condition1 || condition2 || condition3) {
                RuleResult result1 = new RuleResult();
                result1.setCreateTime(clientLog.getDetailTime());
                result1.setRuleId(MockDataUtils.ruleIdGen());
                result1.setEventId(clientLog.getEventId());
                result1.setRuleComplexParam(ruleComplexParam);
                if (condition1) result1.setIsBaseRuleParams(condition1);
                if (condition2) result1.setIsBaseRuleParams(condition2);
                if (condition3) result1.setIsBaseRuleParams(condition3);
                out.collect(result1);
                log.info("基本的规则结果为：{} 组合规则结果为：{} 顺序组合规则结果为：{}",condition1,condition2,condition3);
            }else {
                log.info("不匹配规则{} ",clientLog.getUserNo());
            }
        }
    }
}
